/**
 * Copyright 2019 asiainfo Inc.
 **/
package com.myz.rabbitmq.rpc;

import com.myz.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 * Rpc服务端
 * 1. 配置连接工厂
 * 2. 建立TCP连接
 * 3. 在TCP连接的基础上创建通道
 * 4. 声明一个rpc_queue队列
 * 5. 设置同时最多只能获取一个消息
 * 6. 在rpc_queue队列在等待消息
 * 7. 收到消息后，调用回调对象对消息进行处理，向此消息的replyTo队列中发送处理并带上correlationId
 * 8. 使用wait-notify实现主线程和消息处理回调对象进行同步
 *
 * @author maoyz0621 on 19-1-15
 * @version: v1.0
 */
public class RpcServer {

    private static final Logger logger = LoggerFactory.getLogger(RpcServer.class);
    private static final String RPC_QUEUE_NAME = "rpc_queue";

    public static void execute() throws Exception {
        Connection connection = ConnectionUtils.getConnection();

        try {
            // 在TCP连接的基础上创建通道
            final Channel channel = connection.createChannel();
            // 声明一个rpc_queue队列
            channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
            // 设置同时最多只能获取一个消息
            channel.basicQos(1);
            logger.debug(" [RpcServer] Awaiting RPC requests");

            // 定义消息的回调处理类
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                    // 生成返回的结果，关键是设置correlationId值
                    AMQP.BasicProperties replyProps = new AMQP.BasicProperties
                            .Builder()
                            .correlationId(properties.getCorrelationId())
                            .build();

                    // 生成返回
                    String response = generateResponse(body);
                    // 回复消息，通知已经收到请求
                    channel.basicPublish("", properties.getReplyTo(), replyProps, response.getBytes("UTF-8"));
                    // 对消息进行应答
                    channel.basicAck(envelope.getDeliveryTag(), false);
                    // 唤醒正在消费者所有的线程
                    synchronized (this) {
                        this.notify();
                    }
                }
            };

            // 消费消息
            channel.basicConsume(RPC_QUEUE_NAME, false, consumer);

            // 在收到消息前，本线程进入等待状态
            while (true) {
                synchronized (consumer) {
                    try {
                        consumer.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                // 空值判断，为了代码简洁略
                connection.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 暂停10s，并返回结果
     */
    private static String generateResponse(byte[] body) {
        logger.debug(" [RpcServer] receive requests: " + new String(body));

        try {
            Thread.sleep(1000 * 1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "response:" + new String(body) + "-" + System.currentTimeMillis();
    }
}
